package org.xiaoyu.rpc.seralization.support;

import com.esotericsoftware.kryo.util.Pool;
import io.protostuff.LinkedBuffer;
import io.protostuff.ProtostuffIOUtil;
import io.protostuff.Schema;
import io.protostuff.runtime.RuntimeSchema;
import org.xiaoyu.rpc.seralization.Serializer;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Supplier;

/**
 * Created By Have
 * 2021/7/28 16:13
 */
public class ProtobufSerializer implements Serializer {
    private static final ThreadLocal<LinkedBuffer> bufferLocal = ThreadLocal.withInitial(new Supplier<LinkedBuffer>() {
        @Override
        public LinkedBuffer get() {
            return LinkedBuffer.allocate(10000);
        }
    });
    private Map<Class<?>, Schema<?>> schemaCache = new HashMap<>();

    @Override
    public int getSerializerAlgorithm() {
        return SerializerAlgorithm.PROTOBUF.getType();
    }

    @Override
    public byte[] serializer(Object obj) {
        Class<?> clazz = obj.getClass();
        Schema schema = getSchema(clazz);
        byte[] bytes;
        LinkedBuffer buf = BufferPool.POOL.obtain();
        try {
            bytes = ProtostuffIOUtil.toByteArray(obj, schema, buf);
        } finally {
            buf.clear();
            BufferPool.POOL.free(buf);
        }

        return bytes;
    }

    @Override
    public <T> T deserializer(Class<T> clazz, byte[] bytes) {
        Schema schema = getSchema(clazz);
        Object obj = schema.newMessage();
        ProtostuffIOUtil.mergeFrom(bytes, obj, schema);
        return (T) obj;
    }

    @SuppressWarnings("unchecked")
    public  Schema getSchema(Class clazz) {
        Schema schema = schemaCache.get(clazz);
        if (schema == null) {
            synchronized (schemaCache) {
                if (schemaCache.get(clazz) == null) {
                    schema = RuntimeSchema.getSchema(clazz);
                    schemaCache.put(clazz, schema);
                }
            }
        }
        return schema;
    }

    private static class BufferPool {
        private static final Pool<LinkedBuffer> POOL = new Pool<LinkedBuffer>(true, false, 300) {
            @Override
            protected LinkedBuffer create() {
                return LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
            }
        };
    }

}
